#include "cgzCTcpC.h"
#include "cgzCThreadPool.h"

cgzCLoopC::cgzCLoopC(ETCP_C etype)
    : type(etype)
{
}

cgzCLoopC::~cgzCLoopC()
{
}

ETCP_C cgzCLoopC::getEType() const
{
    return type;
}

cgzCAcptC::cgzCAcptC()
    : cgzCLoopC(TCPC_ACPT)
{
}

cgzCAcptC::~cgzCAcptC()
{
}

cgzCSendC::cgzCSendC()
    : cgzCLoopC(TCPC_SEND),
      m_data(nullptr),
      m_len(0),
      m_sendlen(0)
{
}

// 正常tcp发送消息
cgzCSendC::cgzCSendC(
    const int mid,
    const char *data,
    const int len,
    const long cid)
    : cgzCLoopC(TCPC_SEND),
      m_len(len + sizeof(cgzHeader)),
      m_sendlen(0)
{
    // 拷贝发送数据
    m_data = (char *)malloc(m_len);
    auto &header = *(cgzHeader *)m_data;
    header.mid = mid;
    header.len = len;
    header.all = m_len;
    header.cid = cid;
    memcpy(m_data + sizeof(cgzHeader), data, len);
}

// 网关消息转发
cgzCSendC::cgzCSendC(cgzHeader &header, const int ufd)
    : cgzCLoopC(TCPC_SEND),
      m_len(header.len + sizeof(cgzGateHeader)),
      m_sendlen(0)
{
    // 开辟控制
    m_data = (char *)malloc(m_len);
    // 构造头部
    auto &gateHeader = *(cgzGateHeader *)m_data;
    gateHeader.mid = header.mid;
    gateHeader.len = header.len;
    gateHeader.cid = header.cid;
    gateHeader.all = m_len;
    gateHeader.ufd = ufd;
}

cgzCSendC::~cgzCSendC()
{
    free(m_data);
    m_data = nullptr;
    m_len = 0;
    m_sendlen = 0;
}

cgzCRecvC::cgzCRecvC(int orgid, int tarid, const char *data, const int len)
    : cgzCLoopC(TCPC_RECV),
      m_data((char *)malloc(len)),
      m_len(len)
{
    origin = orgid;
    target = tarid;
    // 拷贝发送数据
    memcpy(m_data, data, len);
}

cgzCRecvC::cgzCRecvC(int orgid, int tarid, const int len)
    : cgzCLoopC(TCPC_RECV),
      m_data((char *)malloc(len)),
      m_len(len)
{
    origin = orgid;
    target = tarid;
}

cgzCRecvC::~cgzCRecvC()
{
    free(m_data);
    m_data = nullptr;
    m_len = 0;
}

cgzCDiscC::cgzCDiscC(int orgid, int tarid)
    : cgzCLoopC(TCPC_DISC)
{
    origin = orgid;
    target = tarid;
}

cgzCDiscC::~cgzCDiscC()
{
}

cgzCGAddC::cgzCGAddC()
    : cgzCLoopC(TCPC_GADD)
{
}

cgzCGAddC::~cgzCGAddC()
{
}

cgzCGDelC::cgzCGDelC()
    : cgzCLoopC(TCPC_GDEL)
{
}

cgzCGDelC::~cgzCGDelC()
{
}

cgzCGClrC::cgzCGClrC()
    : cgzCLoopC(TCPC_GCLR)
{
}

cgzCGClrC::~cgzCGClrC()
{
}

cgzCTcpC::cgzCTcpC(
    cgzCThreadPool *pool,
    const char *name,
    const std::string ip,
    const int port,
    const int cid)
    : cgzITask(pool, name, cgzETask::CTcpC),
      m_ip(ip),
      m_port(port),
      m_cid(cid),
      m_fd(-1),
      m_hand(false),
      m_readCire(READLEN),
      m_sendCire(SENDLEN)
{
}

cgzCTcpC::~cgzCTcpC()
{
    if (m_fd >= 0)
    {
        close(m_fd);
        m_fd = -1;
    }
}

bool cgzCTcpC::tcpinit()
{
    bool ret = tcplink();
    if (ret)
    {
        // 注册服务
        service();
        // 创建事件线程
        m_thread = new std::thread(&cgzCTcpC::tcpread, this);
        m_pool->push_thread(m_thread);
    }
    return ret;
}

bool cgzCTcpC::run(cgzCLoopC *task)
{
    assert(false);
    return false;
}

bool cgzCTcpC::run(int64_t nms)
{
    while (!m_runTask.empty())
    {
        auto task = m_runTask.front();
        m_runTask.pop_front();
        auto loop = static_cast<cgzCLoopC *>(task);
        if (loop->type == TCPC_SEND)
        {
            auto csend = static_cast<cgzCSendC *>(loop);
            if (m_sendCire.push(csend->m_data, csend->m_len))
            {
                csend->~cgzCSendC();
                free(csend);
            }
            else
            {
                assert(false);
            }
        }
        else
        {
            run(loop);
        }
    }

    tcpsend();
    return true;
}

bool cgzCTcpC::tcplink()
{
    m_fd = socket(AF_INET, SOCK_STREAM, 0);
    if (m_fd < 0)
    {
        cgz_this_error("Create socket failed! name:{} ", m_name.c_str());
        return false;
    }

    int yes = 1;
    setsockopt(m_fd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes));
    setsockopt(m_fd, IPPROTO_TCP, TCP_NODELAY, &yes, sizeof(yes)); // 禁用Nagle

    struct sockaddr_in server_addr;
    memset(&server_addr, 0, sizeof(server_addr));
    server_addr.sin_family = AF_INET;
    server_addr.sin_port = htons(m_port);
    server_addr.sin_addr.s_addr = inet_addr(m_ip.c_str());

    // 阻塞连接
    if (::connect(m_fd, (struct sockaddr *)&server_addr, sizeof(server_addr)) < 0)
    {
        close(m_fd);
        cgz_this_error("Connect socket failed! name:{} ", m_name.c_str());
        return false;
    }

    // 设置套接字为非阻塞模式
    // int flags = fcntl(m_fd, F_GETFL, 0);
    // if (flags == -1)
    // {
    //     cgz_this_error("fcntl F_GETFL socket failed! name:{} ", m_name.c_str());
    //     close(m_fd);
    //     return false;
    // }
    // if (fcntl(m_fd, F_SETFL, flags | O_NONBLOCK) == -1)
    // {
    //     cgz_this_error("fcntl F_SETFL socket failed! name:{} ", m_name.c_str());
    //     close(m_fd);
    //     return false;
    // }

    this->connect();
    cgz_this_error("Connect success name:{} ", m_name.c_str());
    return true;
}

bool cgzCTcpC::todoret(cgzIData *dtask)
{
    m_synTask.push_back(dtask);
    return true;
}

bool cgzCTcpC::retTask(cgzIData *cgzTask)
{
    return todoret(cgzTask);
}

bool cgzCTcpC::casBusy()
{
    // 处理返回事件
    std::unique_lock lock(m_rwmutex);
    if (m_synTask.empty())
    {
        return false;
    }
    m_retTask.splice(m_retTask.end(), m_synTask);
    setFree();
    return true;
}

bool cgzCTcpC::tcpsend()
{
    if (m_fd < 0)
    {
        m_sendCire.clear();
        return false;
    }

    while (!m_sendCire.empty())
    {
        auto *ptail = m_sendCire.tail();
        auto ltail = m_sendCire.read();
        ltail = std::min(ltail, ONESEND);
        ssize_t sendlen = send(m_fd, ptail, ltail, 0);
        if (sendlen < 0)
        {
            if (errno == EAGAIN || errno == EWOULDBLOCK)
            {
                return false; // 发送缓冲区满，稍后再试
            }
            else
            {
                m_sendCire.clear();
                disctcp();
                cgz_this_error("Send failed  name:{} err:{}", m_name.c_str(), strerror(errno));
                return false;
            }
        }
        else
        {
            m_sendCire.pop(sendlen);
        }
    }
    return true;
}

bool cgzCTcpC::tcpread()
{
    while (m_bRunning)
    {
        void *readbuf = m_readCire.head();
        int dire = m_readCire.dire();
        if (dire <= 0)
        {
            // 没有接受缓存了区|逻辑错误(非逻辑错误请扩展缓存区)
            cgz_this_error("接收缓存区可能不够 name:{} dire:{}", m_name, dire);
            break;
        }

        int readlen = recv(m_fd, readbuf, dire, 0);
        std::unique_lock lock(m_rwmutex);
        if (readlen < 0)
        {
            if (errno == EAGAIN || errno == EWOULDBLOCK)
            {
                // 读缓冲区空，稍后再试
                break;
            }
            else
            {
                cgz_this_error("Receive failed  name:{} err:{}", m_name.c_str(), strerror(errno));
                disctcp();
                return false;
            }
        }
        else
        {
            m_readCire.push(readlen);
            while (sticking())
                ;
        }
    }

    return true;
}

bool cgzCTcpC::disctcp()
{
    if (m_fd > 0)
    {
        close(m_fd);
        m_fd = -1;
        m_readCire.clear();
        void *node = malloc(sizeof(cgzCDiscC));
        auto discC = new (node) cgzCDiscC(getId(), m_cid);
        todoret(discC);
    }
    return true;
}

bool cgzCTcpC::connect(bool is_hand)
{
    m_hand = is_hand;
    if (is_hand)
    {
        handshake();
    }
    return true;
}
bool cgzCTcpC::handshake()
{
    return sendaccept();
}
bool cgzCTcpC::sendaccept()
{
    auto node = malloc(sizeof(cgzCAcptC));
    auto acptC = new (node) cgzCAcptC();
    acptC->target = m_cid;
    acptC->origin = getId();
    todoret(acptC);
    return true;
}
bool cgzCTcpC::sticking()
{
    cgzHeader header;
    int headLen = sizeof(cgzHeader);
    if (!m_readCire.read((char *)(&header), headLen))
    {
        return false;
    }

    int used = m_readCire.used();
    if (used >= header.all)
    {
        void *node = malloc(sizeof(cgzCRecvC));
        auto readC = new (node) cgzCRecvC(getId(), m_cid, header.all);
        // 消息数据包
        m_readCire.read(readC->m_data, header.all);
        m_readCire.pop(header.all);
        // 将包放入队列
        todoret(readC);
        return true;
    }
    return false;
}
